package sql

import org.apache.spark.SparkConf
import org.apache.spark.sql.expressions.{Aggregator, MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types.{DataType, LongType, MapType, StringType, StructField, StructType}
import org.apache.spark.sql.{Encoder, Encoders, Row, SparkSession, functions}

import java.text.DecimalFormat
import scala.collection.mutable
import scala.collection.mutable.ListBuffer


object E2_TopNV2 {


  """
    |需求： 各区域热门商品Top3
    |
    |我们需要计算各个区域前三大热门商品（这里的热门商品是从点击量的维度来看），并备注上每个商品在主要城市中的分布比例，超过两个城市用其他显示。
    |最终结果类似如下形式：
    |
    |地区	商品名称	点击次数	城市备注
    |华北	商品 A	100000	北京 21.2%，天津 13.2%，其他 65.6%
    |华北	商品 P	80200	北京 63.0%，太原 10%，其他 27.0%
    |华北	商品 M	40000	北京 63.0%，太原 10%，其他 27.0%
    |东北	商品 J	92000	大连 28%，辽宁 17.0%，其他 55.0%
    |""".stripMargin

  def main(args: Array[String]): Unit = {
    topN
  }


  """
    |
    |首先将用户行为数据、产品信息、城市信息三张表进行 JOIN，并筛选出 click_product_id > -1 的记录。将查询结果命名为 t1。
    |接着使用用户自定义的 UDAF（CityRemarkUDAF），实现了对区域内商品点击数量的聚合，并生成城市备注信息。将聚合结果命名为 t2。
    |然后使用 Spark SQL 的窗口函数 rank()，在每个区域内对商品的点击数量进行排名。将排名结果命名为 t3。
    |最后从 t3 表中选择排名在前三位的数据，并显示在控制台上
    |
    |""".stripMargin

  def topN(): Unit = {
    // 创建 Spark 运行配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Hello")
    //创建 SparkSession 对象
    val spark: SparkSession = SparkSession.builder().config(sparkConf)
      .enableHiveSupport() // 添加Hive支持
      .getOrCreate()
    spark.sql("use spark")
    // 查询基础数据
    spark.sql(
      """
        |select
        |a.*,
        |p.product_name,
        |c.area,
        |c.city_name
        |from user_visit_action a
        |join product_info p on a.click_product_id = p.product_id
        |join city_info c on a.city_id = c.city_id
        |where a.click_product_id > -1
        |""".stripMargin
    ).createOrReplaceTempView("t1")
    //根据区域，商品进行数据聚合
    spark.udf.register("cityRemark", new AreaClickUDAF())
    spark.sql(
      """
        |
        |select
        |area,
        |product_name,
        |count(*) as clickCnt,
        |cityRemark(city_name) as city_remark
        |from t1
        |group by area, product_name
        |""".stripMargin
    ).createOrReplaceTempView("t2")
    spark.sql(
      """
        |select
        |*,
        |rank() over(partition by area order by clickCnt desc) as rank
        |from t2
        |""".stripMargin
    ).createOrReplaceTempView("t3")
    spark.sql(
      """
        |select
        |*
        |from t3
        |where rank <=3
        |""".stripMargin
    ).show(false)
    spark.stop()

  }

  class AreaClickUDAF extends UserDefinedAggregateFunction {
    // 输入数据的类型:  北京  String
    override def inputSchema: StructType = {
      StructType(StructField("city_name", StringType) :: Nil)
      //        StructType(Array(StructField("city_name", StringType)))
    }

    // 缓存的数据的类型: 北京->1000, 天津->5000  Map,  总的点击量  1000/?
    override def bufferSchema: StructType = {
      // MapType(StringType, LongType) 还需要标注 map的key的类型和value的类型
      StructType(StructField("city_count", MapType(StringType, LongType)) :: StructField("total_count", LongType) :: Nil)
    }

    // 输出的数据类型  "北京21.2%，天津13.2%，其他65.6%"  String
    override def dataType: DataType = StringType

    // 相同的输入是否应用有相同的输出.
    override def deterministic: Boolean = true

    // 给存储数据初始化
    override def initialize(buffer: MutableAggregationBuffer): Unit = {
      //初始化map缓存
      buffer(0) = Map[String, Long]()
      // 初始化总的点击量
      buffer(1) = 0L
    }

    // 分区内合并 Map[城市名, 点击量]
    override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
      // 首先拿到城市名, 然后把城市名作为key去查看map中是否存在, 如果存在就把对应的值 +1, 如果不存在, 则直接0+1
      val cityName = input.getString(0)
      val map: Map[String, Long] = buffer.getAs[Map[String, Long]](0)
      buffer(0) = map + (cityName -> (map.getOrElse(cityName, 0L) + 1L))
      // 碰到一个城市, 则总的点击量要+1
      buffer(1) = buffer.getLong(1) + 1L
    }

    // 分区间的合并
    override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
      val map1 = buffer1.getAs[Map[String, Long]](0)
      val map2 = buffer2.getAs[Map[String, Long]](0)

      // 把map1的键值对与map2中的累积, 最后赋值给buffer1
      buffer1(0) = map1.foldLeft(map2) {
        case (map, (k, v)) =>
          map + (k -> (map.getOrElse(k, 0L) + v))
      }

      buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)
    }

    // 最终的输出. "北京21.2%，天津13.2%，其他65.6%"
    override def evaluate(buffer: Row): Any = {
      val cityCountMap = buffer.getAs[Map[String, Long]](0)
      val totalCount = buffer.getLong(1)

      var citysRatio: List[CityRemark] = cityCountMap.toList.sortBy(-_._2).take(2).map {
        case (cityName, count) => {
          CityRemark(cityName, count.toDouble / totalCount)
        }
      }
      // 如果城市的个数超过2才显示其他
      if (cityCountMap.size > 2) {
        citysRatio = citysRatio :+ CityRemark("其他", citysRatio.foldLeft(1D)(_ - _.cityRatio))
      }
      citysRatio.mkString(", ")
    }
  }

  case class CityRemark(cityName: String, cityRatio: Double) {
    val formatter = new DecimalFormat("0.00%")
    override def toString: String = s"$cityName:${formatter.format(cityRatio)}"
  }
}
